import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.subjects.BehaviorSubject;

import java.util.Arrays;

/**
 * Created by wangxingsheng on 2019/2/12.
 */
public class RxTest {
    private static Observable<Integer> observable1 = BehaviorSubject.create(emitter -> {
        try {
            Thread.sleep(1000);
            emitter.onNext(1);
        } catch (InterruptedException e) {
            emitter.onError(e);
        } finally {
            emitter.onComplete();
        }
    });

    private static Observable<Integer> observable2 = BehaviorSubject.create(emitter -> {
        try {
            Thread.sleep(2000);
            emitter.onNext(2);
        } catch (InterruptedException e) {
            emitter.onError(e);
        } finally {
            emitter.onComplete();
        }
    });


    public static void main(String[] args) {

        /*
            交叉合并
            observer先后得n个observable发射的结果
            都获取到后，执行onComplete完成方法（observable中必须全部执行了onComplete方法）
         */
        Observable.merge(observable1, observable2)
                //原始Observable没有发射任何数据正常终止（以onCompleted的形式），defaultIfEmpty返回的Observable就发射一个你提供的默认值。
                .defaultIfEmpty(0)
                .subscribe(getObserver("mergeTest"));

        /*
            合并结果
            需要传递一个方法，这个方法用来处理n个observable发射的结果,并返回一个结果
            只要其中一个observable执行了onComplete方法，observer就会执行onComplete完成方法
         */
        Observable.zip(observable1, observable2, (a, b) -> a + b)
                .subscribe(getObserver("zipTest"));

        /*
            将两个发射器合并成一个发射器, 依次发送，发送完一个再接着发送第二个
            observer先后得n个observable发射的结果
            observable执行了onComplete方法，observer才能得到后一个数据源
         */
        Observable.concat(observable1, observable2)
                .subscribe(getObserver("concatTest"));

        /*
            observer只接受首先发送内容的observable
            不管发射的是一项数据还是一个onError或onCompleted通知。amb将忽略和丢弃其它所有observable的发射物
         */
        Observable.amb(Arrays.asList(observable1, observable2))
                .subscribe(getObserver("ambTest"));
    }

    private static Observer<Integer> getObserver(String name) {
        return new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("subscribe " + name);
            }

            @Override
            public void onNext(Integer o) {
                System.out.println(o);
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {
                System.out.println("complete " + name);
            }
        };
    }
}
